package cn.itcast.util;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 设置kafka的生产者对象，作用：发送数据
 */
public class KafkaUtil{

    KafkaProducer kafkaProducer = new KafkaProducer(getProp());
    /**
     * 开发步骤：
     * 1.创建类，泛型参数继承avro基类:指定写入Kafka的数据是Avro类型
     * 2.设置生产者参数
     * 3.自定avro序列化
     * 4.添加发送数据方法
     */

    //2.设置生产者参数
    public Properties getProp(){

        //bootstrap.servers ：broker地址
        //acks ：0,1和-1
        //retries：重试次数
        //batch.size：批量发送大小默认16384 （16kB）
        //linger.ms： 定时发送1ms
        //buffer.memory: 缓存大小33554432
        //key.serializer：key序列化
        //value.serializer： value序列化
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","node1:9092");
        properties.setProperty("acks","1");
        properties.setProperty("retries","0");
        properties.setProperty("batch.size","16384");
        properties.setProperty("linger.ms","1");
        properties.setProperty("buffer.memory","33554432");
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer",StringSerializer.class.getName());
        return properties;
    }

    //4.添加发送数据方法
    public void sendData(String topic ,String data){

        kafkaProducer.send(new ProducerRecord(topic,data));
    }




}
